package Main;

import java.io.IOException;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.ForeachFunction;
import org.apache.spark.sql.Row;
import com.hna.eking.Redis.RedisUtils;
import com.hna.eking.ZkUtils.ZkOptCtx;
import com.hna.eking.ZkUtils.ZkSeqLock;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

public class TestMain {
	
	public static void main(String[] args) {
//		runContext(args[0], args[1], args[3]);
		ZkSeqLock zk = new ZkSeqLock();
		zk.setHosts("10.71.200.109:2181");
		zk.InitWatcher();
		zk.WaitLock();
		zk.CloseWatcher();
		
	}
	public static void RunSql(String SessionID, String UUID, String SqlStr){
		System.out.println("run sql is:" + UUID + "==" + SessionID);
	
	}
	public static void runContext(String SessionID, String UUID, String FileName) {
		System.out.println("run file is:" + UUID + "==" + SessionID);

		SparkConf conf = new SparkConf().setAppName("SparkTest");
		conf.setMaster("spark://127.0.0.1:7702");
		JavaSparkContext sc = new JavaSparkContext(conf);
		
		JavaRDD<String> distFile = sc.textFile("hdfs://127.0.0.1:9001" + FileName);
	
		distFile.cache();
		List<String> Ret = distFile.collect();
		sc.stop();
		
		/*放入redis*/
		RedisUtils redis = new RedisUtils("10.71.200.109", 6379);
		redis.InitJRedis();
		JSONObject json = new JSONObject();
		JSONArray jarry = new JSONArray();
		for(int i = 0; i < Ret.size(); i++){
			jarry.add(i, Ret.get(i));			
		}
		
		json.put("Result", jarry);
		json.put("error", "SUCCESS");
		
		redis.SetStrContext(UUID, json.toJSONString());
		redis.CloseRedis();
		System.out.println("result: " + json.toJSONString());
		
		/*通知zookeeper解锁*/
		try {
			ZkOptCtx zk = new ZkOptCtx();
			zk.setHosts("10.71.200.109:2181");
			zk.setSessionID(SessionID);
			zk.setUUID(UUID);
			zk.SetEnd(ZkOptCtx.PLT_STATUS_SUC);
			zk.CloseZK();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
